package com.zlx.taskcenter.common;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @PackageName: com.goayuan.learn.mq.consumer.modules.redis
 * @ClassName: RedisConsumerContainer
 * @Author: gaoYuan
 * @Date: 2022/2/24 13:28
 * @Description: RedisConsumerContainer 消费者容器
 */
public class RedisConsumerContainer {
    private static final Logger log = LoggerFactory.getLogger(RedisConsumerContainer.class);
    private Map<String, RedisQueueConfiguration> consumerMap = new HashMap<>();
    static boolean run;
    private ExecutorService exec;

    public void addConsumer(RedisQueueConfiguration configuration) {
        Integer num = 1;
        if (null != configuration.getConsumerNum()) {
            num = configuration.getConsumerNum();
        }
        for (int i = 0; i < num; i++) {
            if (consumerMap.containsKey(configuration.getQueue() + i)) {
                log.warn("Key:{} this key already exists, and it will be replaced", configuration.getQueue());
            }
            if (configuration.getConsumer() == null) {
                log.warn("Key:{} consumer cannot be null, this configuration will be skipped", configuration.getQueue());
            }
            consumerMap.put(configuration.getQueue() + i, configuration);
        }
    }

    public void beforeDestroy() {
        run = false;
        this.exec.shutdown();
        log.info("QueueListener exiting.");
        while (!this.exec.isTerminated()) {

        }
        log.info("QueueListener exited.");
    }

    public void init(String threadName) {
        run = true;
        AtomicInteger threadNumber = new AtomicInteger(0);
        this.exec = Executors.newCachedThreadPool(r -> {
            return new Thread(r, threadName + threadNumber.getAndIncrement());
        });
        consumerMap = Collections.unmodifiableMap(consumerMap);
        consumerMap.forEach((k, v) -> exec.submit(new RedisQueueListener(v.getQueue(), v.getConsumer())));
    }
}
